
#include <cstdio>
#include <glib.h>
#include <gio/gio.h>
#include <netinet/in.h>
#include <arpa/inet.h>

//#include "bitrate_estimator.hpp"
#include "codecs.hpp"
#include "h264/h264_nal.hpp"
#include "placeholder.hpp"
#include "rtp.hpp"
#include "xm_app_config.h"
#include "xm_log.h"
#include "xcutil.h"
#include "xrtcp_util.h"
#include "xrtc_endpoint_connector.h"


#define rtcc_dbgd(FMT, ARGS...) dbgd("|%s| " FMT, obj->obj_id, ##ARGS )
#define rtcc_dbgi(FMT, ARGS...) dbgi("|%s| " FMT, obj->obj_id, ##ARGS )

#define PARSING_RTCP_ENABLE 1

struct rtc_connector_st{
    char obj_id[32];
    int auto_free; // free myself when pub rtc destroyed
    rtc_endpoint_t pub_rtc;
    rtc_endpoint_callback_link_t pub_rtc_cb_link;
    GQueue * sub_rtcs;
    rtcp_parser_t pub_rtcp_parser;
    int report_enabled;
    int64_t num_rtp_pkts;
    uint64_t num_fec_pkts;
};

struct rtc_linker_stream_source_info {
    uint32_t local_ssrc;
    uint32_t remb;
};

struct rtc_linker_stream_info {
    bool is_video;
    rtc_linker_stream_source_info sources[2];
};

struct rtc_linker_st{
    rtc_connector_t owner;
    GList * link;
    rtc_endpoint_t rtc;
    rtc_endpoint_callback_link_t rtc_cb_link;
    rtcp_parser_t rtcp_parser;
    bool is_h264_sps_pps_sent = false;
    bool audio;
    bool video;

    bool is_remote;
    char* remote_rtc_id;
    char* remote_addr_cstr;
    GSocket* sock;
    GSocketAddress* addr;
    guint sock_in_source_id;
    int64_t last_rtcp_time;

    rtc_linker_stream_info streams[2];

    int requireCheckRR;

    // 2019.08.21 yangss
    // linker peer ssrcs，貌似现在不起作用
    // 只在处理 sub rtc(local，remote) 终端上行的 rtcp包，而且这个包中的ssrc 已经被替换为了 placeholder
    // 所以跟这个peer ssrc 比较没有任何用途
    std::vector<uint32_t> peer_audio_ssrcs; // 结合 media_service 看，sub rtc 本地时 pub rtc local ssrc；
                                            // 远程时  sub rtc local ssrc
    std::vector<uint32_t> peer_video_ssrcs;
};

static void broadcast_pub_rtp(rtc_linker_t linker, int stream_type,
                              rtc_codec_id_t codec, rtc_codec_id_t fec_codec,
                              int is_rtcp, int is_rtx,
                              const unsigned char* buf, int size);

static void broadcast_pub_rtcp(rtc_linker_t linker, int stream_type,
                               rtc_codec_id_t codec, rtc_codec_id_t fec_codec,
                               int is_rtcp, int is_rtx,
                               const unsigned char* buf, int size);

static void rtc_linker_send_rtp(rtc_linker_t linker, int stream_type,
                         rtc_codec_id_t codec, rtc_codec_id_t fec_codec,
                         bool is_rtx, const unsigned char* buf, int len);

static void rtc_linker_send_rtcp(rtc_linker_t linker, const unsigned char* buf, int len);

static gboolean rtc_linker_recv_remote(GIOChannel *source, GIOCondition cond, gpointer data);

static gboolean rtc_linker_recv_remote_rtcp(rtc_linker_t linker, const unsigned char* buf, int size);

static void rtc_connector_handle_timeout(rtc_connector_t connector, int64_t now_ms);

static inline
unsigned rtc_connector_sub_count_(rtc_connector_t obj){
    if(!obj->sub_rtcs) return 0;
    return g_queue_get_length(obj->sub_rtcs);
}

constexpr int linker_timeout = 60000;

// 销毁linker，如果is_free_list_link=true，将linker从connector删除
// todo: 将is_free_list_link从这里分离出去
static
void destroy_linker(rtc_linker_t linker, int is_free_list_link){
    if(!linker) return ;
    if(linker->rtc && linker->rtc_cb_link){
        rtc_endpoint_unregister_callback(linker->rtc, linker->rtc_cb_link);
        linker->rtc_cb_link = NULL;
        rtc_endpoint_reset_pubrtc(linker->rtc, linker->owner->pub_rtc);
    }
    if(is_free_list_link){
        g_queue_delete_link(linker->owner->sub_rtcs, linker->link);
    }
    if(linker->rtcp_parser){
        rtcp_parser_delete(linker->rtcp_parser);
        linker->rtcp_parser = NULL;
    }

    if (linker->sock_in_source_id > 0) {
        g_source_remove(linker->sock_in_source_id);
    }

    if (linker->sock) {
        GError* err = nullptr;
        g_socket_close(linker->sock, &err);
        if (err != nullptr) {
            g_error_free(err);
        }
        g_object_unref(linker->sock);
        linker->sock = nullptr;
    }

    if (linker->addr) {
        g_object_unref(linker->addr);
        linker->addr = nullptr;
    }

    if (linker->remote_rtc_id) {
        g_free(linker->remote_rtc_id);
        linker->remote_rtc_id = nullptr;
    }

    if (linker->remote_addr_cstr) {
        g_free(linker->remote_addr_cstr);
        linker->remote_addr_cstr = nullptr;
    }

    g_slice_free(struct rtc_linker_st, linker);
}

// connector释放linker列表的回调
static
void on_connector_free_linker(gpointer data){
    rtc_linker_t linker = (rtc_linker_t) data;
    destroy_linker(linker, 0);
}

static
bool is_ssrc_hacked(rtc_linker_t linker, uint32_t ssrc) {
    return (linker->peer_audio_ssrcs.size() > 0 && ssrc == linker->peer_audio_ssrcs[0])
            || (linker->peer_audio_ssrcs.size() > 1 && ssrc == linker->peer_audio_ssrcs[1])
            || (linker->peer_video_ssrcs.size() > 0 && ssrc == linker->peer_video_ssrcs[0])
            || (linker->peer_video_ssrcs.size() > 1 && ssrc == linker->peer_video_ssrcs[1]);
}

/**
 * rtc 本地的，ssrcs is  sub rtc local ssrc == linker.streams[].source.local_ssrc
 *  linker.streams[].source.local_ssrc == sub rtc local ssrc
 *  linker->peer_..._ssrcs == pub_rtc_local_ssrcs
 *  linker --> subC 创建
 * rtc remote. ssrcs is ？？？？
 *  linker.streams[].source.local_ssrc == pub rtc local ssrc
 *  linker->peer_..._ssrcs == sub rtc local ssrc
 *  linker --> pubR 创建
 *
 * @param linker
 * @param nb_ssrcs
 * @param ssrcs
 * @param remb
 * @return
 */
static
bool rtc_linker_update_remb(rtc_linker_t linker, uint8_t nb_ssrcs, const uint32_t* ssrcs, uint32_t remb) {
    bool immediate_feedback = false;
    for (int ssrc_i = 0; ssrc_i < nb_ssrcs && ssrc_i < 4; ssrc_i++) {
        uint32_t ssrc = ssrcs[ssrc_i];

//        dbgi("|%s| rtc_linker_update_remb, ssrc = %u, remb = %u",
//             linker->rtc ? rtc_endpoint_get_id(linker->rtc) : linker->remote_rtc_id, ssrc, remb);

        bool ssrc_match = false;
        if (is_ssrc_hacked(linker, ssrc)) {
            rtc_linker_stream_source_info *source = nullptr;
            if (linker->peer_video_ssrcs.size() > 0 && ssrc == linker->peer_video_ssrcs[0]) {
                //linker->streams[1].sources[0].remb = remb;
                source= &(linker->streams[1].sources[0]);
            } else if (linker->peer_video_ssrcs.size() > 1 && ssrc == linker->peer_video_ssrcs[1]) {
                //linker->streams[1].sources[1].remb = remb;
                source= &(linker->streams[1].sources[1]);
            } else {
                return false;
            }

            if (remb > 0 && remb < source->remb * 0.97) {
                immediate_feedback = true;
            }
            source->remb = remb;

//            dbgi("|%s| rtc_linker_update_remb: set remb, source local ssrc = %u, remb = %u",
//                 linker->rtc ? rtc_endpoint_get_id(linker->rtc) : linker->remote_rtc_id, source->local_ssrc, remb);
        } else {
            rtc_linker_stream_info &video_stream = linker->streams[1];
            for (int source_i = 0; source_i < 2 && !ssrc_match; source_i++) {
                rtc_linker_stream_source_info &source = video_stream.sources[source_i];
                if (source.local_ssrc == ssrc) {
                    // send an remb if new remb is smaller than 97% of old one
                    if (remb > 0 && remb < source.remb * 0.97) {
                        immediate_feedback = true;
                    }
                    source.remb = remb;
                    ssrc_match = true;

//                    dbgi("|%s| rtc_linker_update_remb: set remb, source local ssrc = %u, remb = %u",
//                         linker->rtc ? rtc_endpoint_get_id(linker->rtc) : linker->remote_rtc_id, source.local_ssrc, remb);
                }
            }
        }

    }
    return immediate_feedback;
}

// 处理订阅者发来的rtcp
static inline
void process_sub_rtcp(rtc_connector_t obj, rtc_linker_t linker, int stream_type, unsigned char * buf, int len){
    int pkttype = rtcp_parse_first(linker->rtcp_parser, buf, len);
    const RTCPPacket * body = rtcp_get_body(linker->rtcp_parser);
    int report_nb = 0;
    while( pkttype  != kInvalid){
        switch(pkttype){
            case kPsfbPli:
                rtcc_dbgi("rtc '%s' got pli from subscriber '%s'",
                          rtc_endpoint_get_id(obj->pub_rtc), linker->remote_rtc_id);

                rtc_endpoint_req_pli(obj->pub_rtc);
                pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                break;
                
            case kRtpfbNack:
                pkttype = rtcp_parse_next_item(linker->rtcp_parser);
                while (pkttype == RTCPPacketTypes::kRtpfbNackItem) {
                    rtcc_dbgd("rtc '%s' got nack from subscriber '%s': pid=%u, blp=0x%02X",
                              rtc_endpoint_get_id(obj->pub_rtc), linker->remote_rtc_id,
                              body->NACKItem.PacketID, body->NACKItem.BitMask);
                    rtc_endpoint_req_nack(obj->pub_rtc, stream_type, body->NACKItem.PacketID, body->NACKItem.BitMask);
                    pkttype = rtcp_parse_next_item(linker->rtcp_parser);
                }
                break;

            case kPsfbApp:
                pkttype = rtcp_parse_next_item(linker->rtcp_parser);
                if (pkttype == RTCPPacketTypes::kPsfbRemb) {
                    pkttype = rtcp_parse_next_item(linker->rtcp_parser);
                    if (pkttype == RTCPPacketTypes::kPsfbRembItem) {
                        if (body->REMBItem.NumberOfSSRCs > 0) {
                            bool immediate_feedback = rtc_linker_update_remb(linker,
                                            body->REMBItem.NumberOfSSRCs,
                                            body->REMBItem.SSRCs,
                                            body->REMBItem.BitRate);
                            if (immediate_feedback) {
                                rtc_endpoint_send_immediate_remb_for_video_stream(obj->pub_rtc);
                            }
//                            rtcc_dbgi("rtc '%s' got remb from subscriber '%s': bitrate=%dkbps",
//                                      rtc_endpoint_get_id(obj->pub_rtc), linker->remote_rtc_id,
//                                      body->REMBItem.BitRate / 1000);
//                            for (int i = 0; i < body->REMBItem.NumberOfSSRCs; ++i) {
//                                linker->remb = body->REMBItem.BitRate;
//                                // remote linker 没有rtc
//                                if (!linker->is_remote) {
//                                    rtc_endpoint_update_remb(linker->rtc, body->REMBItem.SSRCs[i], body->REMBItem.BitRate);
//                                }
//                            }
                        }
                    }
                }
                if(pkttype != RTCPPacketTypes::kPsfbRembItem){
                    rtcc_dbgi("rtcp-[name=%s,pkttype=%d]", "unknown-rtcp-sub-remb" , pkttype);
                }
                pkttype =  rtcp_parse_next_packet(linker->rtcp_parser);
                break;
            case kReportBlockItem: {
                uint32_t media_ssrc = body->ReportBlockItem.SSRC;
                uint32_t hseq = body->ReportBlockItem.ExtendedHighestSequenceNumber;
                uint32_t lsr = body->ReportBlockItem.LastSR;
                uint32_t dslr = body->ReportBlockItem.DelayLastSR;
                uint8_t lost_frac = body->ReportBlockItem.FractionLost;
                uint32_t lost_cum = body->ReportBlockItem.CumulativeNumOfPacketsLost;
                uint32_t jitter = body->ReportBlockItem.Jitter;
                int64_t rcv_ts = get_timestamp_ms();
                uint32_t now = ms_to_ntp_mid32(rcv_ts);
                uint32_t rtt_ntp_mid32 = 0;

                if (now > lsr) {
                    rtt_ntp_mid32 = now - lsr;
                } else {
                    // handling wrapped now as follows:
                    // rtt = (32bit_unsigned_max - lsr) + now + 1
                    rtt_ntp_mid32 = 0xFFFFFFFF - lsr + now + 1;
                }
                if (rtt_ntp_mid32 < dslr) {
                    rtcc_dbgi("pubrtc '%s' subscriber(sub '%s'), delay since last report larger then possible, ssrc=%u(?=placeholder::(video|audio).local) now=%u, lsr=%u, rtt_ntp_mid32=%u < dslr=%u",
                              rtc_endpoint_get_id(obj->pub_rtc), linker->is_remote ? linker->remote_rtc_id : rtc_endpoint_get_id(linker->rtc),
                              media_ssrc, now, lsr, rtt_ntp_mid32, dslr);
                    report_nb -= 1;
                    if (report_nb > 0) {
                        pkttype = rtcp_parse_next_item(linker->rtcp_parser);
                    } else {
                        pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                    }
                    continue;
                }
                rtt_ntp_mid32 -= dslr;
                // remote linker 没有 rtc, 目前其report被忽略
                if (!linker->is_remote) {
                    rtc_endpoint_update_report(linker->rtc, rcv_ts, media_ssrc, hseq, rtt_ntp_mid32, jitter, lost_frac, lost_cum);
                }
//                rtcc_dbgi("recv subscriber rtc '%s' rtcp report block ssrc=%X hseq=%u, lsr=%u, rtt=%ums, jitter=%u, lost=%u/%u",
//                          linker->remote_rtc_id, media_ssrc, hseq, lsr, (rtt_ntp_mid32 * 1000) >> 16, jitter, lost_frac, lost_cum);


                report_nb -= 1;
                if (report_nb > 0) {
                    pkttype = rtcp_parse_next_item(linker->rtcp_parser);
                } else {
                    pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                }
            }
            break;
            case kSr:
                if (body->SR.NumberOfReportBlocks > 0) {
                    report_nb = body->SR.NumberOfReportBlocks;
                    pkttype = rtcp_parse_next_item(linker->rtcp_parser);
                } else {
                    pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                }
                break;
            case kRr:
                if (body->RR.NumberOfReportBlocks > 0) {
                    report_nb = body->RR.NumberOfReportBlocks;
                    pkttype = rtcp_parse_next_item(linker->rtcp_parser);
                } else {
                    pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                }
                break;
            case kSdes:
                pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                break;
            case kTransportFeedback:
                {
                    auto fb = rtcp_get_transport_feedback(linker->rtcp_parser);
                    if (!fb) {
                        pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                        break;
                    }
                    if (!linker->is_remote) {
                        rtc_endpoint_update_transport_feedback(linker->rtc, *fb);
                    }
                    pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                }
                break;
            default:
                rtcc_dbgi("rtcp-[name=%s,pkttype=%d]", "unknown-rtcp-sub" , pkttype);
                pkttype = rtcp_parse_next_packet(linker->rtcp_parser);
                break;
        }
    }
}

// 处理subscribe发来数据的回调函数，只处理rtcp，不处理rtp
// 不透传 rtcp时，process_sub_rtcp，处理 sub rtcp；比如将消息转换下，发给 pub_rtc remote（终端）
static
bool on_sub_rtc_stream_data(rtc_endpoint_t sub_rtc, void * context, int stream_type, rtc_codec_id_t codec, rtc_codec_id_t fec_codec, int data_type, int is_rtcp, int is_rtx, unsigned char * buf, int len){
    rtc_linker_t linker = (rtc_linker_t) context;
    rtc_connector_t obj = linker->owner;
    if(data_type == RTC_RECV_CLEAR){
        if(is_rtcp){
            if(obj->report_enabled){
                process_sub_rtcp(obj, linker, stream_type, buf, len);
            }else{
                uint8_t pt = buf[1];
                if (pt == PT_RR) {
                    rtc_endpoint_send_stream_data(obj->pub_rtc, stream_type, codec, fec_codec, is_rtcp, is_rtx, buf, len);
                }
            }
        }
    }

    return false;
}

static
void on_sub_rtc_stream_event(rtc_endpoint_t rtc, void * context, int event, void * event_arg1, int event_arg2){
    rtc_linker_t linker = (rtc_linker_t) context;
    rtc_connector_t obj = linker->owner;
    if(event == RTC_EVENT_SETUP){
        rtcc_dbgi("rtc '%s' subscriber rtc '%s' event: setup",
                  rtc_endpoint_get_id(linker->owner->pub_rtc), rtc_endpoint_get_id(rtc));
    }
}


static
void on_sub_rtc_destroy(rtc_endpoint_t rtc, void * context){
    dbgi("on_sub_rtc_destroy");

    rtc_linker_t linker = (rtc_linker_t) context;
    rtc_connector_t obj = linker->owner;

    rtcc_dbgi("rtc '%s' subscriber rtc '%s' event: destroyed",
              rtc_endpoint_get_id(obj->pub_rtc), rtc_endpoint_get_id(rtc));
    //销毁linker并将其从connector注销
    destroy_linker(linker, 1);
    if (rtc_connector_sub_count_(obj) == 0) {
        rtc_endpoint_enable_report(obj->pub_rtc, 1);
    }

    dbgi("on_sub_rtc_destroy end");
}

// 广播发布者发布的数据
static
void broadcast_pub_data(rtc_connector_t obj, int stream_type, rtc_codec_id_t codec,
                        rtc_codec_id_t fec_codec, int data_type, int is_rtcp, int is_rtx,
                        unsigned char * buf, int len){
    unsigned num_subs = rtc_connector_sub_count_(obj);
    if(num_subs == 0){
        // do nothing
    }
//    else if(num_subs == 1){
//        GList * list = obj->sub_rtcs->head;
//        rtc_linker_t linker = (rtc_linker_t) list->data;
//        rtc_endpoint_send_stream_data(linker->rtc, stream_type, codec, is_rtcp, buf, len);
//    }
    else{
        GList * list = obj->sub_rtcs->head;
        rtcc_dbgd("broadcast_pub_data the list cout is:%d rtcp %d stream type is:%d,%s", num_subs,is_rtcp,stream_type,obj->obj_id);

        while (list){
            GList *next = list->next;
            rtc_linker_t linker = (rtc_linker_t) list->data;
            if (linker->rtc != nullptr && !rtc_endpoint_is_setup(linker->rtc)) {
                rtcc_dbgd("rtc '%s' not setup. when call broadcast_pub_data", rtc_endpoint_get_id(linker->rtc));
                list = next;
                continue;
            }
            if (!is_rtcp) {
                broadcast_pub_rtp(linker, stream_type, codec, fec_codec, is_rtcp, is_rtx, buf, len);
            } else {
                broadcast_pub_rtcp(linker, stream_type, codec, fec_codec, is_rtcp, is_rtx, buf, len);
            }

            list = next;
        }
    }
}

/**
 * sub 本地时：
 *  rtp包的ssrc（pub rtc remote ssrc (终端)） 不被修改，在 通过sub rtc 发送时，直接替换为 sub rtc 的local ssrc （send_rtp_data 替换）
 * sub 远端时：
 *  rtp包的ssrc（pub rtc remote ssrc (终端)），被替换为pub rtc local ssrc；创建linker时，使用pub rtc 的local ssrc 构造了 stream[].source
 *  通过socket，将数据 push 到 remote rtc上
 *
 * @param linker
 * @param stream_type
 * @param codec
 * @param fec_codec
 * @param is_rtcp
 * @param is_rtx
 * @param buf
 * @param size
 */
static void broadcast_pub_rtp(rtc_linker_t linker, int stream_type,
                              rtc_codec_id_t codec, rtc_codec_id_t fec_codec,
                              int is_rtcp, int is_rtx, const unsigned char* buf, int size) {

	rtc_connector_t obj = (rtc_connector_t) linker->owner;
//    rtcc_dbgi("broadcast_pub_rtp step 1 the rtc is '%s' audio:%d the video is %d remote is:%d",
//    		rtc_endpoint_get_id(linker->rtc),linker->audio,linker->video,linker->is_remote);


    if (!linker->audio && stream_type == RTC_AUDIO) {

        return;
    }
    if (!linker->video && stream_type == RTC_VIDEO) {
        return;
    }

    // todo: 研究一下是否有必要发sps/pps，而是只要发送pli即可
//    if (codec == rtc_codec_id_h264 && !linker->is_h264_sps_pps_sent) {
//        int sps_pps_len = rtc_endpoint_get_h264_sps_pps_len(obj->pub_rtc);
//        if (sps_pps_len) {
//            rtcc_dbgi("sending sps pps len=%d", sps_pps_len);
//            memcpy(data, rtc_endpoint_get_h264_sps_pps(obj->pub_rtc), sps_pps_len);
//            rtc_endpoint_send_stream_data(linker->rtc, stream_type, codec, fec_codec, false, false,
//                                          data, sps_pps_len);
//            linker->is_h264_sps_pps_sent = true;
//        }
//    }
    if (linker->is_remote) {
        rtc_linker_send_rtp(linker, stream_type, codec, fec_codec, is_rtx, buf, size);
    } else {
        rtcc_dbgi("broadcast_pub_rtp step 2 the rtc is '%s' audio:%d the video is %d remote is:%d,%d",
        		rtc_endpoint_get_id(linker->rtc),linker->audio,linker->video,linker->is_remote,stream_type);
        int ret = rtc_endpoint_send_stream_data(linker->rtc, stream_type, codec, fec_codec, is_rtcp, is_rtx, (unsigned char*)buf, size);

        if(ret){
            rtc_connector_t obj = (rtc_connector_t) linker->owner;
            rtcc_dbgi("rtc '%s' send stream data, ret = %d", rtc_endpoint_get_id(linker->rtc), ret);
        }
    }

}

/**
 * rtc 本地时，在process_recv_rtp_rtcp方法中，将ssrc（pub rtc remote ssrc (终端)）替换为 placeholder；然后 在 rtc_endpoint_send_stream_data 中替换为sub rtc local ssrc
 * rtc remote时，在process_recv_rtp_rtcp方法中，将ssrc（pub rtc remote ssrc (终端)）替换为 placeholder；然后 通过socket 发到 remote rtc
 *
 * @param linker
 * @param stream_type
 * @param codec
 * @param fec_codec
 * @param is_rtcp
 * @param is_rtx
 * @param buf
 * @param size
 */
static void broadcast_pub_rtcp(rtc_linker_t linker, int stream_type,
                               rtc_codec_id_t codec, rtc_codec_id_t fec_codec,
                               int is_rtcp, int is_rtx, const unsigned char* buf, int size) {
    int pt = buf[1];
    // 不是应该只发送SR才对吗？
    if (pt != PT_RR) {
        if (linker->is_remote) {
            rtc_linker_send_rtcp(linker, buf, size);
        } else {
            rtc_endpoint_send_stream_data(linker->rtc, stream_type, codec, fec_codec, is_rtcp, is_rtx, (unsigned char*) buf, size);
        }
    }
}

/**
 * 当pub、sub在同一个xswitch上时，pub发送来的RTCP 包括sr
 * 这个方法只处理SR
 * 而且只是更新sub rtc transport中的数据，
 *
 * sub rtc的slow_timer --> handle_port向sub rtc 发送sr
 *
 * 这个与 处在另外xswitch上的sub rtc不一样，这些rtc，将会直接被发送sr
 *
 * @param obj
 * @param stream_type
 * @param ntp_msw
 * @param ntp_lsw
 * @param rtp_ts
 */
static inline
void broadcast_pub_ntp(rtc_connector_t obj, int stream_type, uint32_t ntp_msw, uint32_t ntp_lsw, uint32_t rtp_ts){
    GList * list = obj->sub_rtcs->head;
    while (list){
        GList *next = list->next;
        rtc_linker_t linker = (rtc_linker_t) list->data;
        if (linker->is_remote || (linker->rtc != nullptr && !rtc_endpoint_is_setup(linker->rtc))) {
            if(!linker->is_remote){
                rtcc_dbgd("rtc '%s' not setup. when call broadcast_pub_data", rtc_endpoint_get_id(linker->rtc));
            }

            list = next;
            continue;
        }
        rtc_endpoint_set_ntp(linker->rtc, stream_type, ntp_msw, ntp_lsw, rtp_ts);

        list = next;
    }
}

// CNSL  密级数据
// CNSL 只会从pub流中发出，并发送给所有sub流
static inline
void broadcast_pub_CNSL(rtc_connector_t obj, int stream_type, uint8_t l){
    GList * list = obj->sub_rtcs->head;
    while (list){
        GList *next = list->next;
        rtc_linker_t linker = (rtc_linker_t) list->data;
        if (linker->is_remote || (linker->rtc != nullptr && !rtc_endpoint_is_setup(linker->rtc))) {
            if(!linker->is_remote){
                rtcc_dbgd("rtc '%s' not setup. when call broadcast_pub_data", rtc_endpoint_get_id(linker->rtc));
            }

            list = next;
            continue;
        }
        rtc_endpoint_set_CNSL(linker->rtc, stream_type, l);

        list = next;
    }
}
static inline
void process_pub_rtcp(rtc_connector_t obj, int stream_type, rtc_codec_id_t codec,
                      rtc_codec_id_t fec_codec, int data_type, int is_rtcp, int is_rtx,
                      unsigned char * buf, int len){
    int pkttype = rtcp_parse_first(obj->pub_rtcp_parser, buf, len);
    const RTCPPacket * body = rtcp_get_body(obj->pub_rtcp_parser);
    while( pkttype  != kInvalid){
        bool brecv_sr = false;
        //rtcc_dbgi("process_pub_rtcp pub rtc = %s pkttype=%d", rtc_endpoint_get_id(obj->pub_rtc), pkttype);
        switch(pkttype){
            case kSr:
                brecv_sr = true;
                // 下一行注释打开，当pub sub 不再同一个实例上时，就会直接发送给remote xswitch，而且会相同数据会发送两份
                // on_pub_rtc_stream_data 中也发送了一分。所以注释掉。
                // 这样一来，pub sub 不再同一个xswitch上时，pub sr处理，将不一致。
                // 同一个xswitch上时，通过sr，更新sub rtc 数据，通过handle report定时发送 一秒一次。
                // 不在同一个上面时，直接发送
                //broadcast_pub_data(obj, stream_type, codec, fec_codec, data_type, is_rtcp, is_rtx, buf, len);
                break;
            case kRr:
            case kSdes:
            case kRtpfbNack:
            case kPsfbPli:
            case kPsfbApp:
            case kBye:
            case kApp:
                {
                    //CNSL
                    if(1129206604 != body->APP.Name) {
                        break;
                    }
                    if(rtcp_parse_next_item(obj->pub_rtcp_parser) != kAppItem) {
                        break;
                    }
                    // broadcast_pub_data(obj, stream_type, codec, fec_codec, data_type, is_rtcp, is_rtx, buf, len);
                    broadcast_pub_CNSL(obj, stream_type, body->APP.Data[3]);
                }
                break;
            case kTransportFeedback:
                break;
            default:
                rtcc_dbgi("rtcp-[name=%s,pkttype=%d]", "unknown-rtcp-pub" , pkttype);
                break;
        }
        if(brecv_sr) {
            broadcast_pub_ntp(obj, stream_type, body->SR.NTPMostSignificant, body->SR.NTPLeastSignificant, body->SR.RTPTimestamp);
        }
        pkttype = rtcp_parse_next_packet(obj->pub_rtcp_parser);
    }
}

// 这里有问题，本来发送SR和计算RTT是在一个进程内
// 支持远程订阅后，由订阅进程发送SR，由发布进程计算RTT，两个进程的时钟如果不一致，就计算不正确
// 这里发送了所有的RTCP、订阅进程收到SR，就好像从peer收到SR一样，从中取出NTP、下次发送SR时带出去。
// 这里还是需要优化。
static void broadcast_pub_rtcp_to_remote(rtc_connector_t connector, unsigned char* buf, int size) {
    int pkttype = rtcp_parse_first(connector->pub_rtcp_parser, buf, size);
    // const RTCPPacket * body = rtcp_get_body(connector->pub_rtcp_parser);
    while( pkttype != kInvalid){
        switch(pkttype){
            case kSr:
                //dbgi("connector '%s' broadcast SR", connector->obj_id);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
            case kRr:
                //dbgi("connector '%s' broadcast RR", connector->obj_id);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
            case kSdes:
                //dbgi("connector '%s' broadcast SDES", connector->obj_id);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
            case kRtpfbNack:
                //dbgi("connector '%s' broadcast RTPFB NACK", connector->obj_id);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
            case kPsfbPli:
                //dbgi("connector '%s' broadcast PSFB PLI", connector->obj_id);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
            case kPsfbApp:
                //dbgi("connector '%s' broadcast PSFB APP", connector->obj_id);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
            case kBye:
                //dbgi("connector '%s' broadcast BYE", connector->obj_id);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
            case kTransportFeedback:
                //dbgi("connector '%s' broadcast TRANSPORT FEEDBACK", connector->obj_id);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
            default:
                //dbgi("rtcp-[name=%s,pkttype=%d]", "unknown-rtcp-pub" , pkttype);
                pkttype = rtcp_parse_next_packet(connector->pub_rtcp_parser);
                break;
        }
    }

    if (connector->sub_rtcs == nullptr) {
        return;
    }

    GList * list = connector->sub_rtcs->head;
    while (list){
        // GList *next = list->next;
        rtc_linker_t linker = (rtc_linker_t) list->data;

        list = list->next;

        if (!linker->is_remote) {
            continue;
        }

        rtc_linker_send_rtcp(linker, buf, size);
    }
}

static
bool on_pub_rtc_stream_data(rtc_endpoint_t rtc, void * context, int stream_type, rtc_codec_id_t codec, rtc_codec_id_t fec_codec, int data_type, int is_rtcp, int is_rtx, unsigned char * buf, int len){

    rtc_connector_t obj = (rtc_connector_t) context;
    if(data_type == RTC_RECV_CLEAR){
        if(!is_rtcp){
            if(stream_type == RTC_VIDEO){
                if (codec == rtc_codec_id_ulpfec) {
                    ++obj->num_fec_pkts;
                } else {
                    ++obj->num_rtp_pkts;
                }
            }
            broadcast_pub_data(obj, stream_type, codec, fec_codec, data_type, is_rtcp, is_rtx, buf, len);
        }else{
            if(obj->report_enabled){
                process_pub_rtcp(obj, stream_type, codec, fec_codec, data_type, is_rtcp, is_rtx, buf, len);
                broadcast_pub_rtcp_to_remote(obj, buf, len); //通过socket 直接发送到 另外xswitch实例上，只将 SR 发送到了 sub rtc
            }else{ // 多 xswitch 实例时，另外xswitch上的sub不会收到 rtcp包，特别是sr ？？？
                broadcast_pub_data(obj, stream_type, codec, fec_codec, data_type, is_rtcp, is_rtx, buf, len);
            }
        }
    }

    return false;
}

static
void on_pub_rtc_stream_event(rtc_endpoint_t rtc, void * context, int event, void * event_arg1, int event_arg2){
    rtc_connector_t obj = (rtc_connector_t) context;
    if (event == RTC_EVENT_SETUP) {
        rtcc_dbgi("publisher rtc '%s' event: setup", rtc_endpoint_get_id(rtc));
    } else if (event == RTC_EVENT_TIMER) {
        int64_t now_ms = *static_cast<int64_t*>(event_arg1);
        rtc_connector_handle_timeout(obj, now_ms);
    }
}


static
void on_pub_rtc_destroy(rtc_endpoint_t rtc, void * context){
    dbgi("on_pub_rtc_destroy");
    rtc_connector_t obj = (rtc_connector_t) context;
    obj->pub_rtc_cb_link = NULL;
    if(obj->auto_free){
        rtc_connector_delete(obj);
    }
    rtcc_dbgi("publisher rtc '%s' event: destroyed", rtc_endpoint_get_id(rtc));
}



static
rtc_endpoint_callbacks g_pub_rtc_listener = {
    .context = NULL, // context
    .on_destroy = on_pub_rtc_destroy,
    .on_stream_data = on_pub_rtc_stream_data, //  on_stream_data;
    .on_event = on_pub_rtc_stream_event

};

static
rtc_endpoint_callbacks g_sub_rtc_listener = {
    .context = NULL, // context
    .on_destroy = on_sub_rtc_destroy,
    .on_stream_data = on_sub_rtc_stream_data, //  on_stream_data;
    .on_event = on_sub_rtc_stream_event

};

// todo: 删除auto_free这个参数
// rtc_endpoint销毁时，如果有connector就会销毁，所以这里auto_free没有用
rtc_connector_t rtc_connector_create(rtc_endpoint_t pub_rtc, int auto_free){
    int ret = -1;
    rtc_connector_t obj = NULL;


    do {
        obj = (rtc_connector_t)g_slice_alloc0(sizeof(struct rtc_connector_st));
        snprintf(obj->obj_id, sizeof(obj->obj_id), "conn-%s", rtc_endpoint_get_id(pub_rtc));
        obj->pub_rtc = pub_rtc;
        obj->auto_free = auto_free;
        obj->pub_rtc_cb_link = rtc_endpoint_register_callback(obj->pub_rtc, obj, &g_pub_rtc_listener);
        obj->pub_rtcp_parser = rtcp_parser_create();
        obj->report_enabled = app_config_get()->feedback_passthrough ? 0 : 1;
        rtc_endpoint_enable_report(obj->pub_rtc, obj->report_enabled);
        ret = 0;
    } while (0);

    if(ret){
        rtc_connector_delete(obj);
        obj = NULL;
    }

    return obj;
}

void rtc_connector_delete(rtc_connector_t obj){
    if(!obj) return;

    rtc_endpoint_enable_report(obj->pub_rtc, 1);

    if(obj->sub_rtcs){
        //        g_slist_free_full(rtc->sub_rtcs, (GDestroyNotify)&g_free);
        g_queue_free_full(obj->sub_rtcs, (GDestroyNotify) on_connector_free_linker);
        obj->sub_rtcs = NULL;
    }

    if(obj->pub_rtc && obj->pub_rtc_cb_link){
        rtc_endpoint_unregister_callback(obj->pub_rtc, obj->pub_rtc_cb_link);
        obj->pub_rtc_cb_link = NULL;
    }

    if(obj->pub_rtcp_parser){
        rtcp_parser_delete(obj->pub_rtcp_parser);
        obj->pub_rtcp_parser = NULL;
    }

    g_slice_free(struct rtc_connector_st, obj);
}

void rtc_linker_setup_stream_info(rtc_linker_t linker, rtc_endpoint_t rtc) {
    linker->streams[0].is_video = false;
    linker->streams[1].is_video = true;

    std::vector<uint32_t> audio_ssrcs;
    std::vector<uint32_t> video_ssrcs;
    rtc_endpoint_get_local_audio_ssrc(rtc, audio_ssrcs);
    rtc_endpoint_get_local_video_ssrc(rtc, video_ssrcs);

    for (size_t i = 0; i < audio_ssrcs.size() && i < 2; ++i) {
        linker->streams[0].sources[i].local_ssrc = audio_ssrcs[0];
    }
    for (size_t i = 0; i < video_ssrcs.size() && i < 2; ++i) {
        linker->streams[1].sources[i].local_ssrc = video_ssrcs[i];
    }
}

rtc_linker_t rtc_connector_add_sub(rtc_connector_t obj, rtc_endpoint_t rtc, bool audio, bool video){
    // TODO: check if rtc is obj->pub_rtc
    if(!obj->sub_rtcs){
        obj->sub_rtcs = g_queue_new();
    }

    rtc_linker_t linker = (rtc_linker_t)g_slice_alloc0 (sizeof(struct rtc_linker_st));
    linker->owner = obj;
    linker->rtc = rtc;
    g_queue_push_tail(obj->sub_rtcs, linker);
    linker->link = g_queue_peek_tail_link(obj->sub_rtcs);
    linker->rtc_cb_link = rtc_endpoint_register_callback(rtc, linker, &g_sub_rtc_listener);
    linker->rtcp_parser = rtcp_parser_create();
    linker->audio = audio;
    linker->video = video;
    linker->is_h264_sps_pps_sent = false;
    linker->remote_rtc_id = g_strdup(rtc_endpoint_get_id(rtc));

    std::vector<uint32_t> peer_audio_ssrcs;
    std::vector<uint32_t> peer_video_ssrcs;
    rtc_endpoint_get_local_audio_ssrc(rtc, peer_audio_ssrcs);
    rtc_endpoint_get_local_video_ssrc(rtc, peer_video_ssrcs);

    linker->peer_audio_ssrcs = peer_audio_ssrcs; // pubR时这里时 sub local ssrc；所以为了一致，在rtcmagr调用时，使用了rtc
    linker->peer_video_ssrcs = peer_video_ssrcs;
    rtcc_dbgi("rtc %s, linker peer ssrcs [(%u, %u), (%u, %u)]",
              rtc_endpoint_get_id(linker->rtc),
              linker->peer_audio_ssrcs.size() > 0 ? linker->peer_audio_ssrcs[0] : 0,
              linker->peer_audio_ssrcs.size() > 1 ? linker->peer_audio_ssrcs[1] : 0,
              linker->peer_video_ssrcs.size() > 0 ? linker->peer_video_ssrcs[0] : 0,
              linker->peer_video_ssrcs.size() > 1 ? linker->peer_video_ssrcs[1] : 0);

    rtc_linker_setup_stream_info(linker, obj->pub_rtc); //pubR时 linker.stream[].sources[].local_ssrc = pub local ssrc

    rtc_endpoint_enable_report(linker->rtc, obj->report_enabled);
    if (video) {
        rtc_endpoint_req_pli(obj->pub_rtc);
    }
    return linker;
}

void rtc_connector_update_sub(rtc_connector_t obj, rtc_linker_t linker, bool audio, bool video) {
    linker->audio = audio;
    linker->video = video;
    if (video) {
        rtc_endpoint_req_pli(obj->pub_rtc);
    }
    return;
}

int rtc_connector_remove_remote_sub(rtc_connector_t connector, const std::string& sub_rtc_id) {
    if (connector->sub_rtcs == nullptr) {
        return 0;
    }
    auto it = connector->sub_rtcs->head;
    while (it != nullptr) {
        auto linker = static_cast<rtc_linker_t>(it->data);
        it = it->next;
        if (linker->is_remote && strcmp(linker->remote_rtc_id, sub_rtc_id.c_str()) == 0) {
            destroy_linker(linker, 1);
            break;
        }
    }

    if (rtc_connector_sub_count_(connector) == 0) {
        rtc_endpoint_enable_report(connector->pub_rtc, 1);
    }

    return 0;
}

int rtc_connector_remove_sub(rtc_connector_t obj, rtc_linker_t linker){
    if (!obj || !linker) {
        return -1;
    }
    destroy_linker(linker, 1);
    if (rtc_connector_sub_count_(obj) == 0) {
        rtc_endpoint_enable_report(obj->pub_rtc, 1);
    }
    return 0;
}

unsigned rtc_connector_sub_count(rtc_connector_t obj){
    if (obj == nullptr) return 0;
    return rtc_connector_sub_count_(obj);
}

void rtc_connector_each_linker(rtc_connector_t obj, std::function<void(rtc_linker_t)> func) {
    if (obj == nullptr) {
        return;
    }
    GList * head = obj->sub_rtcs->head;
    while (head){
        func(static_cast<rtc_linker_t>(head->data));
        head = head->next;
    }
}

rtc_endpoint_t rtc_linker_get_rtc(rtc_linker_t linker) {
    if (linker == nullptr) return nullptr;
    return linker->rtc;
}

/**
 * After a second thought, I found another way to implement remote subscription.
 * Which is:
 * - implement an new endpoint class to represent a remote one
 * - in the case of local side, an old rtc_endpoint will forward all received
 *   packet to remote side
 * - in the case of remote side, when packet received, it acts like an old
 *   rtc_endpoint, which means, it can be a connector, with local linkers
 * todo: design the shadow rtc pattern, then compare it with current one
 *   if it is better, implement it.
 */
rtc_linker_t rtc_connector_find_linker_by_address(rtc_connector_t connector,
    const std::string remote_addr) {
    if (connector->sub_rtcs == nullptr) {
        return nullptr;
    }
    GList * list = connector->sub_rtcs->head;
    while (list){
        rtc_linker_t linker = (rtc_linker_t) list->data;
        list = list->next;

        if (!linker->is_remote) {
            continue;
        }

        if (strcmp(remote_addr.c_str(), linker->remote_addr_cstr) == 0) {
            return linker;
        }
    }

    return nullptr;
}

int rtc_connector_add_remote_sub(rtc_connector_t connector, const std::string &sub_rtc_id,
                                 bool audio, bool video, const std::string& link_addr,
                                 int requireCheckRR, std::vector<uint32_t> sub_peer_audio_ssrcs, std::vector<uint32_t> sub_peer_video_ssrcs) {

    // find if linker already exist
    // if exists, just update it with audio/video flags
    rtc_linker_t old_linker = rtc_connector_find_linker_by_address(connector,
            link_addr);
    if (old_linker != nullptr) {
        old_linker->audio = audio;
        old_linker->video = video;
        old_linker->last_rtcp_time = get_timestamp_ms();
        dbgi("connector '%s' updated existing remote sub '%s': audio=%s, video=%s",
             rtc_endpoint_get_id(connector->pub_rtc),
             sub_rtc_id.c_str(),
             audio ? "true" : "false", video ? "true" : "false");
        return 0;
    }

    GSocketAddress* addr = nullptr;
    auto host_port_sep = link_addr.find(":");
    if (host_port_sep == std::string::npos || host_port_sep == link_addr.length()) {
        dbge("connector '%s' add remote sub '%s' error: could not parse link addr %s",
             rtc_endpoint_get_id(connector->pub_rtc), sub_rtc_id.c_str(), link_addr.c_str());
        return 400;
    }

    auto host = link_addr.substr(0, host_port_sep);
    auto port = atoi(link_addr.substr(host_port_sep+1).c_str());
    addr = g_inet_socket_address_new_from_string(host.c_str(), port);
    if (addr == nullptr) {
        dbge("connector '%s' add remote sub '%s' error: could not convert link addr %s to GSocketAddress",
             rtc_endpoint_get_id(connector->pub_rtc), sub_rtc_id.c_str(), link_addr.c_str());
        return 400;
    }

    GError* err = nullptr;
    GSocket* sock = g_socket_new(G_SOCKET_FAMILY_IPV4, G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err);
    if (err != nullptr) {
        dbge("connector '%s' add remote sub '%s' error: could not create remote link (sender) socket: %s",
             rtc_endpoint_get_id(connector->pub_rtc), sub_rtc_id.c_str(), err->message);
        g_error_free(err);
        err = nullptr;
        return 500;
    }

    g_socket_bind(sock, g_inet_socket_address_new(g_inet_address_new_any(G_SOCKET_FAMILY_IPV4), 0), FALSE, &err);
    if (err != nullptr) {
        dbge("connector '%s' add remote sub '%s' error: could not bind socket: %s",
             rtc_endpoint_get_id(connector->pub_rtc), sub_rtc_id.c_str(), err->message);
        g_error_free(err);
        err = nullptr;
        g_socket_close(sock, &err);
        if (err != nullptr) {
            g_error_free(err);
        }
        g_object_unref(sock);
        return 500;
    }

    if(connector->sub_rtcs == nullptr){
        connector->sub_rtcs = g_queue_new();
    }

    rtc_linker_t linker = (rtc_linker_t)g_slice_alloc0 (sizeof(struct rtc_linker_st));
    linker->owner = connector;
    linker->rtc = nullptr;
    g_queue_push_tail(connector->sub_rtcs, linker);
    linker->link = g_queue_peek_tail_link(connector->sub_rtcs);
    linker->rtcp_parser = rtcp_parser_create();
    linker->audio = audio;
    linker->video = video;
    linker->is_h264_sps_pps_sent = false;
    linker->is_remote = true;
    linker->remote_rtc_id = g_strdup(sub_rtc_id.c_str());
    linker->remote_addr_cstr = g_strdup(link_addr.c_str());
    linker->sock = sock;
    linker->addr = addr;
    linker->requireCheckRR = requireCheckRR;


    linker->streams[0].is_video = false;
    std::vector<uint32_t> audioSsrcs;
    rtc_endpoint_get_relay_audio_ssrc_list(connector->pub_rtc, audioSsrcs);
    //linker->streams[0].sources[0].local_ssrc = 754039198;
    linker->streams[0].sources[0].local_ssrc = audioSsrcs[0];

    linker->streams[1].is_video = true;
    std::vector<uint32_t> videoSsrcs;
    rtc_endpoint_get_relay_video_ssrc_list(connector->pub_rtc, videoSsrcs);
    linker->streams[1].sources[0].local_ssrc = videoSsrcs[0]; //pub rtc local ssrc
    linker->streams[1].sources[1].local_ssrc = videoSsrcs[1];

    linker->peer_audio_ssrcs = sub_peer_audio_ssrcs; // 结合 media_service 看，这个时 sub rtc local ssrc
    linker->peer_video_ssrcs = sub_peer_video_ssrcs;
//    dbgi("rtc %s, linker peer ssrcs [(%u, %u), (%u, %u)]",
//         linker->remote_rtc_id,
//         linker->peer_audio_ssrcs[0], linker->peer_audio_ssrcs[1],
//         linker->peer_video_ssrcs[0], linker->peer_video_ssrcs[1]);

    if (linker->peer_audio_ssrcs.size() > 0) {
        dbgi("rtc %s, linker peer audio ssrcs0=%u",
             linker->remote_rtc_id, linker->peer_audio_ssrcs[0]);
    }
    if (linker->peer_audio_ssrcs.size() > 1) {
        dbgi("rtc %s, linker peer audio ssrcs1=%u",
             linker->remote_rtc_id, linker->peer_audio_ssrcs[1]);
    }
    if (linker->peer_video_ssrcs.size() > 0) {
        dbgi("rtc %s, linker peer video ssrcs0=%u",
             linker->remote_rtc_id, linker->peer_video_ssrcs[0]);
    }
    if (linker->peer_video_ssrcs.size() > 1) {
        dbgi("rtc %s, linker peer video ssrcs1=%u",
             linker->remote_rtc_id, linker->peer_video_ssrcs[1]);
    }

    auto fd = g_socket_get_fd(sock);
    auto ch = g_io_channel_unix_new(fd);
    linker->sock_in_source_id = g_io_add_watch(ch, G_IO_IN, rtc_linker_recv_remote, linker);
    g_io_channel_unref(ch);

    linker->last_rtcp_time = get_timestamp_ms();

    if(linker->is_remote && !linker->requireCheckRR){
        dbgw("linker remote '%s' - '%s'. require checkRR:%d xswitch update timestamp. after pub rtp",
             linker->remote_rtc_id, linker->remote_addr_cstr, linker->requireCheckRR);
    }

    return 0;
}

static void rtc_linker_send_rtp(rtc_linker_t linker, int stream_type, rtc_codec_id_t codec,
                         rtc_codec_id_t fec_codec, bool is_rtx, const unsigned char* buf, int len) {
    // 因为要在发送之前，修改rtp内容，所以要先复制
    unsigned char obuf[1600] = {0,};

    uint8_t x = buf[0] >> 4 & 1;
    uint8_t cc = buf[0] & 0x0f;
    uint16_t header_size = 12 + cc * 4;
    uint16_t header_extension_size = 0;

    if (x == 1) {
        header_extension_size = be_get_u16(buf + header_size + 2);
    }

    if (len < header_size + header_extension_size) {
        dbge("remote link could not send packet: not enough data");
        return;
    }

    memcpy(obuf, buf, len);

    uint8_t pt = rtc_codec_default_pt(codec, fec_codec);
    obuf[1] &= 0x80;
    obuf[1] |= pt;

    if (stream_type == RTC_AUDIO && !is_rtx) {
        //be_set_u32(placeholder::rtp_audio_ssrc, obuf + 8);
        be_set_u32(linker->streams[0].sources[0].local_ssrc, obuf + 8); //pub rtc local ssrc
    } else if (stream_type == RTC_AUDIO && is_rtx) {
        //be_set_u32(placeholder::rtp_audio_rtx_ssrc, obuf + 8);
        be_set_u32(linker->streams[0].sources[1].local_ssrc, obuf + 8); //pub rtc local ssrc
    } else if (stream_type == RTC_VIDEO && !is_rtx) {
        //be_set_u32(placeholder::rtp_video_ssrc, obuf + 8);
        be_set_u32(linker->streams[1].sources[0].local_ssrc, obuf + 8); //pub rtc local ssrc
    } else if (stream_type == RTC_VIDEO && is_rtx) {
        //be_set_u32(placeholder::rtp_video_rtx_ssrc, obuf + 8);
        be_set_u32(linker->streams[1].sources[1].local_ssrc, obuf + 8); //pub rtc local ssrc
    }


    GError* err = nullptr;

//    sockaddr_in sa;
//    g_socket_address_to_native(linker->addr, &sa, sizeof(sa), &err);
//
//    dbgi("send to %s:%d", inet_ntoa(sa.sin_addr), ntohs(sa.sin_port));

    // int bytes = g_socket_send_to(linker->sock, linker->addr, (gchar*)obuf,
    //                              len, nullptr, &err);
    g_socket_send_to(linker->sock, linker->addr, (gchar*)obuf,
                                len, nullptr, &err);
    if (err != nullptr) {
        dbge("remote link could not send rtp packet: %s", err->message);
        return;
    }

    if(linker->is_remote && !linker->requireCheckRR){
        //平安旁路接受服务器，没有发出RR，导致超时被移除
        //不需要检查接受端的RR时，直接更新last time，这样，不会导致 未收到RR而超时被移除
        linker->last_rtcp_time = get_timestamp_ms();
//        dbge("linker remote '%s' - '%s'. require checkRR:%d xswitch update timestamp. after pub rtp",
//             linker->remote_rtc_id, linker->remote_addr_cstr, linker->requireCheckRR);
    }
}

static void rtc_linker_send_rtcp(rtc_linker_t linker, const unsigned char* buf, int len) {
    GError* err = nullptr;
    g_socket_send_to(linker->sock, linker->addr, (gchar*)buf,
                     len, nullptr, &err);
    if (err != nullptr) {
        dbge("remote link could not send rtcp packet: %s", err->message);
        return;
    }
}

/**
 * 远端 sub rtc
 * pub rtc --rtc|rtcp-> socket --rtp|rtcp-> sub rtc(remote)
 * sub rtc(remote) --rtcp-> socket --rtcp-> pub rtc
 *
 * 此方法 将处理 这个 sub rtc(remote) 发送过来的 rtcp包
 *
 * 本地 sub rtc，由 on_sub_rtc_stream_data 处理。
 *
 * @param source
 * @param cond
 * @param data
 * @return
 */
static gboolean rtc_linker_recv_remote(GIOChannel *source, GIOCondition cond, gpointer data) {
    auto linker = static_cast<rtc_linker_t>(data);

    // auto fd = g_socket_get_fd(linker->sock);

    unsigned char buf[1600] = {0,};

    GError* err = nullptr;

    int size = g_socket_receive_from(linker->sock, nullptr, (gchar*)buf, sizeof(buf), nullptr, &err);

    gboolean status = G_SOURCE_CONTINUE;

    if (is_rtcp(buf, size)) {
        status = rtc_linker_recv_remote_rtcp(linker, buf, size);
    }

    return status;
}

//gboolean rtc_linker_recv_remote_rtp(rtc_linker_t linker, const unsigned char* buf, int size) {
//    if (linker->rtc == nullptr) {
//        return TRUE;
//    }
//    auto ssrc = be_get_u32(buf + 8);
//
//    int stream_type;
//    bool is_rtx;
//
//    if (ssrc == placeholder::rtp_audio_ssrc) {
//        stream_type = RTC_AUDIO;
//        is_rtx = false;
//    } else if (ssrc == placeholder::rtp_audio_rtx_ssrc) {
//        stream_type = RTC_AUDIO;
//        is_rtx = true;
//    } else if (ssrc == placeholder::rtp_video_ssrc) {
//        stream_type = RTC_VIDEO;
//        is_rtx = false;
//    } else if (ssrc == placeholder::rtp_video_rtx_ssrc) {
//        stream_type = RTC_VIDEO;
//        is_rtx = true;
//    } else {
//        dbgi("could not handle received remote rtp: unknown ssrc '%u'", ssrc);
//        return TRUE;
//    }
//
//    uint8_t pt = buf[1] & 0x7f;
//    rtc_codec_id_t codec = rtc_codec_id_unknown;
//    rtc_codec_id_t fec_codec = rtc_codec_id_unknown;
//
//    switch (pt) {
//    case rtc_codec_default_pt_vp8:
//        codec = rtc_codec_id_vp8;
//        break;
//    case rtc_codec_default_pt_h264:
//        codec = rtc_codec_id_h264;
//        break;
//    case rtc_codec_default_pt_pcmu:
//        codec = rtc_codec_id_pcmu;
//        break;
//    case rtc_codec_default_pt_opus:
//        codec = rtc_codec_id_opus;
//        break;
//    case rtc_codec_default_pt_red:
//        codec = rtc_codec_id_red;
//        break;
//    case rtc_codec_default_pt_ulpfec_vp8:
//        codec = rtc_codec_id_ulpfec;
//        fec_codec = rtc_codec_id_vp8;
//        break;
//    case rtc_codec_default_pt_ulpfec_h264:
//        codec = rtc_codec_id_ulpfec;
//        fec_codec = rtc_codec_id_h264;
//        break;
//    case rtc_codec_default_pt_ulpfec_unknown:
//        codec = rtc_codec_id_ulpfec;
//        fec_codec = rtc_codec_id_unknown;
//        break;
//    default:
//        return TRUE;
//    }
//
//    rtc_endpoint_send_stream_data(linker->rtc, stream_type, codec, fec_codec, 0, is_rtx ? 1 : 0, const_cast<unsigned char*>(buf), size);
//    return TRUE;
//}

static gboolean rtc_linker_recv_remote_rtcp(rtc_linker_t linker, const unsigned char* buf, int size) {
    auto ssrc = rtcp_get_ssrc(buf, size);

    int stream_type;

    if ((linker->peer_audio_ssrcs.size() > 0 && ssrc == linker->peer_audio_ssrcs[0]) //sub rtc local ssrc
        || (linker->peer_audio_ssrcs.size() > 1 && ssrc == linker->peer_audio_ssrcs[1])
        || ssrc == placeholder::audio.remote.pri
        || ssrc == placeholder::audio.remote.rtx) {
        stream_type = RTC_AUDIO;
    } else if ((linker->peer_video_ssrcs.size() > 0 && ssrc == linker->peer_video_ssrcs[0])
               || (linker->peer_video_ssrcs.size() > 1 && ssrc == linker->peer_video_ssrcs[1])
               || ssrc == placeholder::video.remote.pri
               || ssrc == placeholder::video.remote.rtx) {
        stream_type = RTC_VIDEO;
    } else {
        dbgi("could not handle received remote rtcp: unknown ssrc '%u'", ssrc);
        return G_SOURCE_CONTINUE;
    }

    process_sub_rtcp(linker->owner, linker, stream_type, const_cast<unsigned char*>(buf), size);

    linker->last_rtcp_time = get_timestamp_ms();
    return G_SOURCE_CONTINUE;
}

static void rtc_connector_handle_timeout(rtc_connector_t connector, int64_t now_ms) {
    if (connector->sub_rtcs == nullptr) {
        return;
    }
    GList * list = connector->sub_rtcs->head;
    while (list){
        rtc_linker_t linker = (rtc_linker_t) list->data;
        list = list->next;

        if (linker->is_remote) {
            auto elapsed_ms = now_ms - linker->last_rtcp_time;
            if (elapsed_ms > linker_timeout) {
                time_t t = linker->last_rtcp_time / 1000;

                auto ts = localtime(&t);

                char tstr[64] = {0,};

                strftime(tstr, sizeof(tstr), "%a %Y-%m-%d %H:%M:%S %Z", ts);

                dbgi("connector '%s' remote linker '%s' timeout, last seen at '%s'", connector->obj_id, linker->remote_rtc_id, tstr);

                rtc_connector_remove_remote_sub(connector, linker->remote_rtc_id);
            }
            continue;
        }
    }
}

uint32_t rtc_linker_get_video_remb(rtc_linker_t linker) {
    return linker->streams[1].sources[0].remb;
}

